package com.lecosa.flink

import org.apache.flink.api.common.eventtime.WatermarkStrategy
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema
import com.ververica.cdc.connectors.mysql.source.MySqlSource
import com.ververica.cdc.connectors.mysql.table.StartupOptions
import org.apache.flink.streaming.api.scala._

object FlinkCdc2 {


  def createCDCSource(): MySqlSource[String] = {
    val startPos: StartupOptions = if (true) StartupOptions.latest() else StartupOptions.initial()

    val prop = new java.util.Properties()
    prop.setProperty("decimal.handling.mode", "string")

    MySqlSource.builder[String]
      .hostname("110.41.6.107")
      .port(33306)
      .username("root")
      .password("root")
      .databaseList("scm_kaifa")
      .tableList("scm_kaifa.scf_payment_info")
      .startupOptions(startPos)
//      .serverId(params.serverId)
      .debeziumProperties(prop)
      .deserializer(new JsonDebeziumDeserializationSchema)
//      .deserializer(new CustomDeserializationSchema)
      .build()
  }




  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val value = env.fromSource(createCDCSource(), WatermarkStrategy.noWatermarks(), "mysql cdc source")

//      .addSink(new ChangeSinkJson).name("kinesis sink")
//      .setParallelism(2)

    value.print()
    env.execute("MySQL Binlog CDC Kinesis")

  }
}
